package org.chen.service.impl;

import com.google.common.collect.Lists;
import feign.FeignException;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.chen.config.BatchStatsManager;
import org.chen.enu.BatchStatus;
import org.chen.enu.TaskStatus;
import org.chen.exception.ErrorContext;
import org.chen.feign.CommentServiceClient;
import org.chen.mapper.CommentTaskMapper;
import org.chen.model.CommentData;
import org.chen.model.CommentTask;
import org.chen.request.CommentRequest;
import org.chen.response.CommentResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

@Service
@Slf4j
public class CommentTaskExecutionService {

    private static final int PAGE_SIZE = 100; // 每次查询100条

    private static final int PROCESS_BATCH_SIZE = 20; // 每批处理20条

    @Autowired
    private CommentServiceClient commentServiceClient;

    @Autowired
    private CommentErrorHandlerService commentErrorHandlerService;

    @Qualifier("batchTaskExecutor")
    @Autowired
    private AsyncTaskExecutor taskExecutor;

    @Autowired
    private CommentTaskMapper taskMapper;

    @Autowired
    private BatchStatsManager batchStatsManager;

    /**
     * 处理批量任务
     */
    public void processCommentBatch(String batchNo) {
        try {
            batchStatsManager.updateBatchStatus(batchNo, BatchStatus.PROCESSING);

            LocalDateTime lastProcessTime = null;
            int emptyCount = 0;

            while (true) {
                // 检查剩余任务数
                long remainingTasks = taskMapper.countPendingTasks(batchNo);
                if (remainingTasks == 0) {
                    log.info("所有任务处理完成，批次：{}", batchNo);
                    break;
                }

                // 查询待处理任务
                List<CommentTask> pageTasks = taskMapper.selectPendingTasksByTime(
                        batchNo,
                        TaskStatus.PENDING.name(),
                        lastProcessTime,
                        PAGE_SIZE
                );

                if (pageTasks.isEmpty()) {
                    emptyCount++;
                    if (emptyCount >= 3) {
                        if (remainingTasks > 0) {
                            log.warn("批次{}还有{}个任务未完成，重置时间戳重新扫描",
                                    batchNo, remainingTasks);
                            lastProcessTime = null;  // 重置时间戳，重新扫描
                            emptyCount = 0;
                            Thread.sleep(1000);
                            continue;
                        }
                        break;
                    }
                    Thread.sleep(1000);
                    continue;
                }

                // 重置空计数
                emptyCount = 0;

                // 更新最后处理时间
                lastProcessTime = pageTasks.get(pageTasks.size() - 1).getCreateTime();

                // 将任务分成小批次处理
                List<List<CommentTask>> batchList = Lists.partition(pageTasks, PROCESS_BATCH_SIZE);

                // 使用CompletableFuture并发处理小批次
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                for (List<CommentTask> batch : batchList) {
                    CompletableFuture<Void> future = CompletableFuture.runAsync(
                            () -> processBatch(batch),
                            taskExecutor
                    ).exceptionally(throwable -> {
                        log.error("批次处理异常: {}", throwable.getMessage());
                        return null;
                    });
                    futures.add(future);
                }
                // 等待当前页的所有批次处理完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

            }

            // 最终检查和状态更新
            batchStatsManager.finalizeBatchStatus(batchNo);

        } catch (Exception e) {
            log.error("批次执行错误: {}, 错误: {}", batchNo, e.getMessage());
            batchStatsManager.updateBatchStatus(batchNo, BatchStatus.FAILED);
        }
    }

    private void processBatch(List<CommentTask> tasks) {
        for (CommentTask task : tasks) {
            try {
                attemptComment(task);
            } catch (Exception e) {
                log.error("执行错误: taskId={}, error={}",
                        task.getId(), e.getMessage());
            }
        }
    }

    /**
     * 单次评论尝试
     */
    public void attemptComment(CommentTask task) {
        try{
            log.info("开始执行评论:{}", task.getTargetUrl());
            CommentData data = buildCommentData(task);
            // 构建请求对象
            CommentRequest request = CommentRequest.builder()
                    .targetUrl(task.getTargetUrl())
                    .commentData(data)
                    .build();

            // 直接调用并处理响应
            CommentResponse response = commentServiceClient.postComment(request);

            if (response.isSuccess()) {
                handleSuccess(task, response.getRedirectUrl());
            } else {
                // 直接使用评论服务返回的错误上下文
                commentErrorHandlerService.handleTaskError(task, response.getErrorContext());
            }
        }catch (FeignException e) {
            ErrorContext errorContext = commentErrorHandlerService.analyzeFeignError(e);
            commentErrorHandlerService.handleTaskError(task, errorContext);
        } catch (Exception e) {
            commentErrorHandlerService.handleTaskError(task, commentErrorHandlerService.analyzeError(e));
        }
    }

    /**
     * 构建评论数据
     */
    private CommentData buildCommentData(CommentTask task) {
        return CommentData.builder()
                .content(task.getCommentContent())
                .email(task.getEmail())
                .author(task.getNickname())
                .url(task.getWebsiteUrl())
                .build();
    }

    /**
     * 处理成功响应
     */
    private void handleSuccess(CommentTask task, String redirectUrl) {
        try {
            if (redirectUrl != null && !redirectUrl.isEmpty()) {
                taskMapper.updateNewUrl(task.getId(), redirectUrl, LocalDateTime.now());
                task.setStatus(TaskStatus.SUCCESS.name());
                task.setNewCommentUrl(redirectUrl);
                taskMapper.updateById(task);

                batchStatsManager.incrementCounter(task.getBatchNo(), true);
            } else {
                ErrorContext errorContext = ErrorContext.fatalError("未获取到重定向URL");
                commentErrorHandlerService.handleTaskError(task, errorContext);
            }
        } catch (Exception e) {
            ErrorContext errorContext = ErrorContext.fatalError("更新任务状态失败: " + e.getMessage());
            commentErrorHandlerService.handleTaskError(task, errorContext);
        }
    }
}
